import concurrent.futures
import pandas as pd
import os
from datetime import datetime,timedelta,time,timezone
from dateutil.relativedelta import relativedelta
from cc.appcoin2.strategy.lm.kline_generator2 import convert_to_datetime, query_inverval_feed
from coin2.service.feed.realtime_mid_price_service_pb2 import \
    QueryMidPriceRequestProto
from xunkemgmt_client.client.api_client import XunkemgmtClient
from coin2.service.feed.interval_feed_service_pb2 import QueryIntervalFeedRequestProto


os.environ["TZ"] = "GMT"
proto_topic = ['timestamp', 'close_mid',]

def calculate_next_number(symbol):
    for n in range(4, -1, -1):
        if ('N'*n+'EXT' in symbol):
            return n
    return 0

def to_absolute_week_date(date,next_num):
  date = date + timedelta(weeks=next_num)
  days_to_friday = (4 - date.weekday()) % 7
  return date + timedelta(days=days_to_friday)

def to_absolute_month_date(date,next_num):
  date = date + relativedelta(months=next_num+1)
  last_day_of_month = datetime(date.year, date.month, 1) - timedelta(days=1) 
  days_to_friday = (7 - (4 - last_day_of_month.weekday()) % 7) %7
  return last_day_of_month - timedelta(days=days_to_friday)

def to_absolute_quarter_date(date,next_num):
  date = date + relativedelta(months=next_num*3)
  quarter = (date.month - 1) // 3 + 1
  last_month_of_quarter = quarter * 3
  return to_absolute_month_date(datetime(date.year,last_month_of_quarter,1),0)

def to_absolute_day_date(date,next_num):
  return date + timedelta(days=next_num)
  
def covert_norm_symbol(date,norm_symbol:str):
  split_symbol = norm_symbol.split('.')
  if len(split_symbol) > 1 and split_symbol[1] != "PERPETUAL" and not split_symbol[1].isnumeric():
    next_num = calculate_next_number(split_symbol[1])
    if "WEEK" in split_symbol[1]:
      absolute_date = to_absolute_week_date(date,next_num)
    elif "MONTH" in split_symbol[1]:
      absolute_date = to_absolute_month_date(date,next_num)
    elif "QUARTER" in split_symbol[1]:
      absolute_date = to_absolute_quarter_date(date,next_num)
    elif "DAY" in split_symbol[1]:
      absolute_date = to_absolute_day_date(date,next_num)
    absolute_date = absolute_date.strftime('%Y%m%d')
    norm_symbol = norm_symbol.replace(split_symbol[1],absolute_date)
  return norm_symbol
  
def get_symbol_map(date, norm_symbols):
  symbol_map = {}
  for symbol in norm_symbols:
    absolute_symbol = covert_norm_symbol(date,symbol)
    symbol_map[absolute_symbol] = symbol
  return symbol_map

def create_interval_feed_query_proto(start_dt,end_dt,mea,symbols):
  query_arg = QueryIntervalFeedRequestProto(
    start_time=start_dt,
    end_time=end_dt,
    mea=mea,
    symbols=symbols,
    resolution='1m',
    topics=["close_mid"]
    )
  return query_arg

def create_mid_price_provider_query_proto(start_dt,end_dt,mea,symbols):
  query_arg = QueryMidPriceRequestProto(
    start_time=start_dt,
    end_time=end_dt,
    mea=mea,
    symbols=symbols,
    )
  return query_arg

def query_mid_price_provider_feed(request_proto):
  with XunkemgmtClient() as client:
    response = client.query_mid_price(request_proto)
  return response

def generate_request(mea, symbols, start_dt, end_dt, create_func=create_interval_feed_query_proto):
  query_list = []
  symbol_map_list = []
  duration = (end_dt - start_dt).days
  if duration > 0:
    while start_dt < end_dt:
      next_day_dt = datetime.combine((start_dt + timedelta(days=1)).date(),time(0))
      symbol_map = get_symbol_map(start_dt,symbols)
      query_arg = create_func(
          int(start_dt.replace(tzinfo=timezone.utc).timestamp()*1e9),
          int(min(next_day_dt, end_dt).replace(tzinfo=timezone.utc).timestamp()*1e9),
          mea,
          symbol_map.keys()
      )
      start_dt = next_day_dt
      query_list.append(query_arg)
      symbol_map_list.append(symbol_map)
  else:
    symbol_map_list.append(get_symbol_map(start_dt,symbols))
    query_list.append(create_func(
        int(start_dt.replace(tzinfo=timezone.utc).timestamp()*1e9),
        int(end_dt.replace(tzinfo=timezone.utc).timestamp()*1e9),
        mea,
        symbol_map_list[0].keys()
    ))
  return query_list,symbol_map_list

def convert_response_to_df(response,symbol_map):
  interval_feed_df = {}
  if response.status == "success":
    for absolute_symbol, interval_feed in response.result.items():
      interval_feed_map = {}
      for topic in proto_topic:
        feed_data = interval_feed.__getattribute__(topic)
        if len(feed_data)>0:
          interval_feed_map[topic] = list(feed_data)
      interval_feed_df[symbol_map[absolute_symbol]] = pd.DataFrame(interval_feed_map)
  return interval_feed_df

def response_to_df(req_list,fs,symbol_map_list,source):
  errors = []
  oneday_mid_price_df = []
  latest_ts = 0
  for n in range(len(fs)):
    result = fs[n].result()
    mid_price_df = convert_response_to_df(result,symbol_map_list[n])
    if len(mid_price_df) != 0:
      oneday_mid_price_df.append(mid_price_df)
      if len(result.error_message) > 0:
        errors.append(f"EXCEPTION: {req_list[n].start_time}-{req_list[n].end_time}, {result.error_message}")
    else:
      if(source == 'historical'):
        latest_ts = fs[n-1].result().latest_timestamp
        break
      errors.append(f"ERROR: {req_list[n].start_time}-{req_list[n].end_time}, {result.error_message}")
  for err in errors:
    print(err)

  columns_name_map = {"close_mid": "mid_price",}
  symbol_mid_price_df = {}
  failed_symbol = []
  norm_symbols = symbol_map_list[0].values()
  for symbol in norm_symbols:
    df_list = [df[symbol] for df in oneday_mid_price_df if symbol in df.keys()]
    if len(df_list) == 0:
      print(f"exception: {symbol} have no data found in {source} h5 file")
      failed_symbol.append(symbol)
      continue
    symbol_df = pd.concat(df_list).reset_index()
    del symbol_df['index']
    symbol_df = symbol_df.rename(columns_name_map, axis='columns')
    symbol_mid_price_df[symbol] = symbol_df
  return symbol_mid_price_df,latest_ts


def get_mid_price_from_interval_feed(start_dt, end_dt, mea, symbols):
  req_list, symbol_map_list = generate_request(
    mea=mea,
    symbols=symbols,
    start_dt=start_dt,
    end_dt=end_dt
    )
  fs = []
  with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    for req in req_list:
      future = executor.submit(query_inverval_feed, req)
      fs.append(future)
  return response_to_df(req_list,fs,symbol_map_list,'historical')


def get_mid_price_from_provider(start_dt, end_dt, mea, symbols):
  req_list, symbol_map_list = generate_request(
    mea=mea,
    symbols=symbols,
    start_dt=start_dt,
    end_dt=end_dt,
    create_func=create_mid_price_provider_query_proto
    )
  fs = []
  with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    for req in req_list:
      future = executor.submit(query_mid_price_provider_feed, req)
      fs.append(future)
  return response_to_df(req_list,fs,symbol_map_list,'realtime')
  

def get_mid_price(start_dt,end_dt,mea,symbols):
  start_dt = convert_to_datetime(start_dt)
  end_dt = convert_to_datetime(end_dt)
  mid_price_p1, latest_ts= get_mid_price_from_interval_feed(start_dt,end_dt,mea,symbols)
  if(mid_price_p1 and latest_ts == 0):
    return mid_price_p1
  elif(not mid_price_p1 and latest_ts == 0):
    mid_price_p2, _ = get_mid_price_from_provider(start_dt,end_dt,mea,symbols)
    return mid_price_p2
  else:
    provider_query_start = datetime.fromtimestamp(latest_ts/1e9)
    mid_price_p2, latest_ts = get_mid_price_from_provider(provider_query_start,end_dt,mea,symbols)
    mid_price = {}
    for symbol in symbols:
      if symbol in mid_price_p1 and symbol in mid_price_p2:
        mid_price[symbol] = pd.concat([mid_price_p1[symbol],mid_price_p2[symbol]],axis=0)
      elif symbol in mid_price_p1:
        mid_price[symbol] = mid_price_p1[symbol]
      else:
        mid_price[symbol] = mid_price_p2[symbol]
    return mid_price
  
# usage 
# get_mid_price(start,end,mea,symbols)
# start: query start time, use yyyyMMdd’T’HHmmss format eg.20230520T1300
# end: query end time, same format as start
# mea: market.exchange.api_version
# symbols: symbols you want to query. must be a list. either absolute or norm is available. not support option '.ALL' strike price
# example
if __name__ == "__main__":
  mp = get_mid_price('20230613','20230614', 'Options.Deribit.v2', ["BTC-USD.WEEK.C.26000.1","BTC-USD.NEXT_WEEK.C.26000.1"])
  print(mp["BTC-USD.WEEK.C.26000.1"].dropna(how='any'))
  print(mp["BTC-USD.NEXT_WEEK.C.26000.1"].dropna(how='any'))
#   mp2 = get_mid_price('20230518T1200','20230526', 'Options.Deribit.v2', ["BTC-USD.20230519.C.26000.1","BTC-USD.20230602.C.26000.1"])
#   print(mp2["BTC-USD.20230519.C.26000.1"].dropna(how='any'))
#   print(mp2["BTC-USD.20230602.C.26000.1"].dropna(how='any'))